package MapReduce.Demo2_PhoneTrafficCount;

import MapReduce.writableBean.PhoneTrafficWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import utils.JobSubmit;

import java.io.IOException;

/**
 * @Author lixinlei
 * @Date 2023/3/16 9:10
 */
public class PhoneTrafficCountApp {

    public static class PTCMapper extends Mapper<LongWritable, Text,Text, PhoneTrafficWritable>{

        Text outKey = new Text();
        PhoneTrafficWritable outValue = new PhoneTrafficWritable();


        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, PhoneTrafficWritable>.Context context) throws IOException, InterruptedException {

            String line = value.toString();

            //把每行数据按 \t 进行分割，获取到所有字段的数组
            String[] fields = line.split("\t");

            //获取有用字段的数据，并根据实际情况转型
            String phoneNum = fields[1];
            int upFlow = Integer.parseInt(fields[fields.length-3]);
            int downFlow = Integer.parseInt(fields[fields.length-2]);

            //用手机号作为输出的 key
            outKey.set(phoneNum);
            //用上行下行和总流量作为输出的 value (自定义序列化类型)
            outValue.setUpFlow(upFlow);
            outValue.setDownFlow(downFlow);
            outValue.setSumFlow(upFlow+downFlow);


            context.write(outKey,outValue);

        }
    }

    public static class PTCReducer extends Reducer<Text, PhoneTrafficWritable,Text, PhoneTrafficWritable>{

        PhoneTrafficWritable outValue = new PhoneTrafficWritable();

        /**
         * 数据的格式
         *      手机号，(PhoneTrafficWritable,PhoneTrafficWritable,...)
         * @param key
         * @param values
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void reduce(Text key, Iterable<PhoneTrafficWritable> values, Reducer<Text, PhoneTrafficWritable, Text, PhoneTrafficWritable>.Context context) throws IOException, InterruptedException {

            //初始化上行下行和总流量的总和
            int upCountFlow = 0;
            int downCountFlow = 0;
            int sumCountFlow = 0;

            //遍历每个手机号的所有流量数据，分别对不同流量进行相加求和
            for (PhoneTrafficWritable value : values) {
                upCountFlow += value.getUpFlow();
                downCountFlow += value.getDownFlow();
                sumCountFlow += value.getSumFlow();
            }

            outValue.setUpFlow(upCountFlow);
            outValue.setDownFlow(downCountFlow);
            outValue.setSumFlow(sumCountFlow);

            context.write(key,outValue);

        }
    }

    public static void main(String[] args) {
        JobSubmit.submitBaseJob(PhoneTrafficCountApp.class,args);
    }

}
